package org.yzh.client.netty;

import io.github.yezhihao.netmc.core.model.Message;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioChannelOption;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yzh.client.protocol.JT808Beans;
import org.yzh.protocol.basics.JTMessage;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author yezhihao
 * @home https://gitee.com/yezhihao/jt808-server
 */
public class TCPClient {

    private static final Logger log = LoggerFactory.getLogger(TCPClient.class);

    private ClientConfig config;

    private EventLoopGroup workerGroup;

    private Channel channel;

    private Bootstrap bootstrap;

    public TCPClient(ClientConfig config) {
        this.config = config;
    }

    private void startInternal() {
        try {
            TCPClient that = this;
            this.workerGroup = new NioEventLoopGroup();
            bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(workerGroup);
            bootstrap.option(NioChannelOption.SO_REUSEADDR, true).option(NioChannelOption.TCP_NODELAY, true)
                .option(NioChannelOption.SO_KEEPALIVE, true).handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                public void initChannel(NioSocketChannel ch) {
                    ch.pipeline().addLast("frameDecoder",
                        new DelimiterBasedFrameDecoder(config.maxFrameLength, Unpooled.wrappedBuffer(config.delimiter),
                            Unpooled.wrappedBuffer(config.delimiter, config.delimiter)))
                        .addLast("decoder", new ByteToMessageDecoder() {
                            @Override
                            protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf buf,
                                List<Object> out) {
                                Object message = config.decoder.decode(buf);
                                if (message != null)
                                    out.add(message);
                                buf.skipBytes(buf.readableBytes());
                            }
                        }).addLast("encoder", new MessageToByteEncoder<Message>() {
                        @Override
                        protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) {
                            ByteBuf buf = config.encoder.encode(msg);
                            log.info("<<<<<原始报文[ip={}],hex={}", ctx.channel().remoteAddress(),
                                ByteBufUtil.hexDump(buf));
                            out.writeBytes(config.delimiter).writeBytes(buf).writeBytes(config.delimiter);
                        }
                    }).addLast("adapter", new TCPClientHandler(config.handlerMapping, that));
                }
            });

            this.connect();
        } catch (Exception e) {
            log.error("===TCP Client异常关闭", e);
        }
    }

    public void writeObject(Object message) {
        if (channel != null) {
            channel.writeAndFlush(message);
            log.info("<<<<<<<<<<发送消息:{}", message);
        } else {
            log.warn("=== Channel is NULL");
        }
    }

    public synchronized TCPClient start() {
        startInternal();
        log.warn("===TCP Client启动成功, port={}===", config.port);
        return this;
    }

    public synchronized void stop() {
        workerGroup.shutdownGracefully();
        log.warn("===TCP Client已经停止, port={}===", config.port);
    }

    public void connect() {
        if (channel != null && channel.isActive()) {
            return;
        }

        /**
         * 这里connect后添加sync表示如果失败,将重新抛出异常
         */
        ChannelFuture channelFuture = bootstrap.connect(config.ip, config.port);
        channelFuture.addListener((ChannelFutureListener)future -> {
            if (future.isSuccess()) {
                channel = future.channel();
                channel.closeFuture();
                /* 发送注册消息 */
                JTMessage register = JT808Beans.T0100(config.deviceId, config.plateNo);
                register.setHeader(JT808Beans.H2013Header(register.getClass().getAnnotation(
                    io.github.yezhihao.protostar.annotation.Message.class), config.mobileNo));
                this.writeObject(register);

            } else {
                future.channel().eventLoop().schedule(new Runnable() {
                    @Override
                    public void run() {
                        log.info("连接失败,重试中...");
                        connect();
                    }
                }, 5, TimeUnit.SECONDS);
            }

        });
    }
}